﻿#include "MarketDataManager.h"
#include "FileOperator.h"
#include "Tools.h"
#include <algorithm>
#include <iostream>
#include <fstream>
#include <iterator>
#include <string>
#include <vector>
#include <thread>
#include <stdlib.h>
#include "CTPTime.h"
#include "FrameData.h"
#ifdef __linux__
#include <cstring>
#endif
#include "MarketTimeAmendment.h"


MarketDataManager::MarketDataManager()
{
	m_ctpDisconnectTime = 0;
	m_lossSharedMemaryTimes = 0;
}

MarketDataManager::~MarketDataManager()
{
}

void MarketDataManager::cacheFrameData(FData::FrameKItem &frameItem)
{
	std::lock_guard<std::mutex> plock(m_pricesItemLock);
	m_frameItems.emplace_back(frameItem);
}

void MarketDataManager::createTestFrameData()
{
	MData::PriceItem priceItem;
	priceItem.LastPrice = 10.0f;
	priceItem.UpdateTime = Tools::GetCurrentTimeStamp();
	priceItem.UpdateMillisec = 0;
	MData::KLineItem kItem;
	kItem.Open = kItem.High = kItem.Low = priceItem.LastPrice;
	kItem.Close = 0;
	kItem.UpdateTime = priceItem.UpdateTime;
	kItem.UpdateMillisec = 0;
	//FData::FrameKItem framKItem;
	//framKItem.KItem = kItem;
	//framKItem.priceItem = priceItem;
	//cacheFrameData("cu2110", framKItem);
}

bool MarketDataManager::checkCTPDistanceLong()
{
	if (m_ctpDisconnectTime > 0)
	{
		// 获得当前时间
		time_t cuTime = Tools::GetCurrentTimeStamp();
		if (cuTime - m_ctpDisconnectTime > 10)
		{
			return true;
		}
	}
	return false;
}

void MarketDataManager::LoopWait()
{
	//createTestFrameData();
	MDIO::FrameData dataStream;
	int marketEmptyTimes = 0; // 记录行情空的此时
	for (;;)
	{
		// 判断ctp是否断开时间太长了
		if (checkCTPDistanceLong())
		{
			break;
		}
		// 价格数组
		std::list<FData::FrameKItem> tmpPriceItems;
		{
			std::lock_guard<std::mutex> plock(m_pricesItemLock);
			tmpPriceItems.swap(m_frameItems);
		}
		
		if (tmpPriceItems.empty())
		{
			//std::cout << "休眠500ms" << std::endl;
			std::this_thread::sleep_for(std::chrono::milliseconds(500));
			//std::cout << "休眠结束" << std::endl;
			dataStream.ReleaseData();
			++marketEmptyTimes;
		}
		else
		{
			//KIndexType msStamp = Tools::GetCurrentMSTimeStamp();
			marketEmptyTimes = 0;
			bool saveSuc = dataStream.Save(tmpPriceItems);
			if (saveSuc)
			{
				m_lossSharedMemaryTimes = 0;
			}
			else
			{
				++m_lossSharedMemaryTimes;
			}
		}
		if (m_lossSharedMemaryTimes > 10)
		{
			break;
		}
		// 行情空40次，大约20s，那么证明是当前时间段内行情结束了，要做一次全局保存
		if (marketEmptyTimes > 40)
		{
			SaveTheLastKItem();
			marketEmptyTimes = 0;
		}
	}
}

void MarketDataManager::SetCTPDisconnectTime(time_t pTime)
{
	if (m_ctpDisconnectTime <= 0)
	{
		m_ctpDisconnectTime = pTime;
	}
}

void MarketDataManager::ConnectedCTP()
{
	m_ctpDisconnectTime = 0;
}

void MarketDataManager::SaveTheLastKItem()
{
	for (std::map<std::string, KGen::KItemGenerator>::iterator it = m_kLines.begin(); it != m_kLines.end(); ++it)
	{
		const MData::KLineItem &pKItem = it->second.GetLastKItem();
		if (pKItem.UpdateTime > 0)
		{
			m_kItemStream.AppendKItem(it->first, pKItem);
			it->second.ClearKItem();
		}
	}
}

void MarketDataManager::UpdateMarket(const std::string &pSymbol, const MData::PriceItem &priceItem)
{
	// 本地保存价格
	m_priceStream.AppPrice(MarketTimeAmendment::Instance().GetTodayStr(), pSymbol, priceItem);
	
	// 合成K线
	FData::FrameKItem pframKItem;
	MData::KLineItem lastKItem;
	UpdateLocalKItem(pSymbol, priceItem, lastKItem);
	pSymbol.copy(pframKItem.SymbolName, pSymbol.length());
	pframKItem.SymbolName[pSymbol.length()] = 0;
	pframKItem.priceItem = priceItem;
	pframKItem.LastKItemGenTime = lastKItem.UpdateTime;
	cacheFrameData(pframKItem);
}

void MarketDataManager::UpdateLocalKItem(const std::string &pSymbol, const MData::PriceItem &priceItem, MData::KLineItem &lastKItem)
{
	KGen::KItemGenerator &kGenerator = m_kLines[pSymbol];
	//MData::KLineItem lastNewKItem;
	kGenerator.PushNewPrice(priceItem, lastKItem);
	// 在这里保存分钟K线
	if (lastKItem.UpdateTime > 0)
	{
		m_kItemStream.AppendKItem(pSymbol, lastKItem);
	}
}
